/*-------------------------------------------------------------------------
 *
 * inv_api.c
 *	  处理反转文件系统大对象的例程。此文件包含用户级大对象应用程序接口
 *	  例程。
 *
 *
 * 注意：我们使用其 C 结构声明访问 pg_largeobject.data。这是安全的，因为它
 * 紧跟在 pageno 之后，pageno 是一个 int4 字段，因此数据字段总是按 4 字节对齐，
 * 即使它是短 1 字节头格式。我们必须进行解压，因为它很可能是被压缩或短格式。
 * 我们还需要检查 NULL，因为 initdb 将标记 loid 和 pageno，但不标记 data 为
 * NOT NULL。
 *
 * 注意：许多这些例程在 CurrentMemoryContext 中泄漏内存，确实大多数后台代码
 * 也会如此。我们期望 CurrentMemoryContext 将是一个短期上下文。必须在函数
 * 调用之间保持的数据显示在 CacheMemoryContext（关系结构）中或在 inv_open
 * 指定的内存上下文中（针对 LargeObjectDesc 结构）。
 *
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 *
 * IDENTIFICATION
 *	  src/backend/storage/large_object/inv_api.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include <limits.h>

#include "access/detoast.h"
#include "access/genam.h"
#include "access/htup_details.h"
#include "access/sysattr.h"
#include "access/table.h"
#include "access/xact.h"
#include "catalog/dependency.h"
#include "catalog/indexing.h"
#include "catalog/objectaccess.h"
#include "catalog/pg_largeobject.h"
#include "catalog/pg_largeobject_metadata.h"
#include "libpq/libpq-fs.h"
#include "miscadmin.h"
#include "storage/large_object.h"
#include "utils/acl.h"
#include "utils/fmgroids.h"
#include "utils/rel.h"
#include "utils/snapmgr.h"


/*
 * GUC：向后兼容标志以抑制 LO 权限检查
 */
bool		lo_compat_privileges;

/*
 * 所有对 pg_largeobject 及其索引的访问都使用单个 Relation 引用，
 * 这样我们每个事务只需打开 pg_relation 一次。
 * 为了避免当第一次此类引用发生在子事务内时出现问题，
 * 我们执行稍微笨拙的手段，将 Relation 引用的所有权分配给 TopTransactionResourceOwner。
 */
static Relation lo_heap_r = NULL;
static Relation lo_index_r = NULL;


/*
 * 打开 pg_largeobject 及其索引，如果当前 xact 中尚未完成
 */
static void fc_open_lo_relation(void)
{
	ResourceOwner fc_currentOwner;

	if (lo_heap_r && lo_index_r)
		return;					/* 当前事务中已经打开 */

	/* 安排顶部事务拥有这些关系引用 */
	fc_currentOwner = CurrentResourceOwner;
	CurrentResourceOwner = TopTransactionResourceOwner;

	/* 使用 RowExclusiveLock，因为我们可能会读取或写入 */
	if (lo_heap_r == NULL)
		lo_heap_r = table_open(LargeObjectRelationId, RowExclusiveLock);
	if (lo_index_r == NULL)
		lo_index_r = index_open(LargeObjectLOidPNIndexId, RowExclusiveLock);

	CurrentResourceOwner = fc_currentOwner;
}

/*
 * 在主事务结束时进行清理
 */
void close_lo_relation(bool fc_isCommit)
{
	if (lo_heap_r || lo_index_r)
	{
		/*
		 * 只有在提交时才需要关闭；否则，由中止清理来处理
		 * 它
		 */
		if (fc_isCommit)
		{
			ResourceOwner fc_currentOwner;

			fc_currentOwner = CurrentResourceOwner;
			CurrentResourceOwner = TopTransactionResourceOwner;

			if (lo_index_r)
				index_close(lo_index_r, NoLock);
			if (lo_heap_r)
				table_close(lo_heap_r, NoLock);

			CurrentResourceOwner = fc_currentOwner;
		}
		lo_heap_r = NULL;
		lo_index_r = NULL;
	}
}


/*
 * 与 pg_largeobject.c 的 LargeObjectExists() 相同，除了可以
 * 指定用于读取的快照。
 */
static bool fc_myLargeObjectExists(Oid fc_loid, Snapshot fc_snapshot)
{
	Relation	pg_lo_meta;
	ScanKeyData fc_skey[1];
	SysScanDesc fc_sd;
	HeapTuple	fc_tuple;
	bool		fc_retval = false;

	ScanKeyInit(&fc_skey[0],
				Anum_pg_largeobject_metadata_oid,
				BTEqualStrategyNumber, F_OIDEQ,
				ObjectIdGetDatum(fc_loid));

	pg_lo_meta = table_open(LargeObjectMetadataRelationId,
							AccessShareLock);

	fc_sd = systable_beginscan(pg_lo_meta,
							LargeObjectMetadataOidIndexId, true,
							fc_snapshot, 1, fc_skey);

	fc_tuple = systable_getnext(fc_sd);
	if (HeapTupleIsValid(fc_tuple))
		fc_retval = true;

	systable_endscan(fc_sd);

	table_close(pg_lo_meta, AccessShareLock);

	return fc_retval;
}


/*
 * 从 pg_largeobject 元组中提取数据字段，必要时进行解压
 * 并验证长度是否合理。返回数据指针（一个 bytea *），
 * 数据长度，以及是否需要释放数据指针的指示。
 */
static void fc_getdatafield(Form_pg_largeobject fc_tuple,
			 bytea **fc_pdatafield,
			 int *fc_plen,
			 bool *fc_pfreeit)
{
	bytea	   *fc_datafield;
	int			fc_len;
	bool		fc_freeit;

	fc_datafield = &(fc_tuple->data); /* 请参阅文件顶部的说明 */
	fc_freeit = false;
	if (VARATT_IS_EXTENDED(fc_datafield))
	{
		fc_datafield = (bytea *)
			detoast_attr((struct varlena *) fc_datafield);
		fc_freeit = true;
	}
	fc_len = VARSIZE(fc_datafield) - VARHDRSZ;
	if (fc_len < 0 || fc_len > LOBLKSIZE)
		ereport(ERROR,
				(errcode(ERRCODE_DATA_CORRUPTED),
				 errmsg("pg_largeobject entry for OID %u, page %d has invalid data field size %d",
						fc_tuple->loid, fc_tuple->pageno, fc_len)));
	*fc_pdatafield = fc_datafield;
	*fc_plen = fc_len;
	*fc_pfreeit = fc_freeit;
}


/*
 *	inv_create -- 创建一个新的大对象
 *
 *	参数：
 *	  lobjId - 用于新大对象的 OID，或 InvalidOid 随机选择一个
 *
 *	返回：
 *	  新对象的 OID
 *
 * 如果 lobjId 不是 InvalidOid，则当 OID 已被使用时会发生错误。
 */
Oid inv_create(Oid fc_lobjId)
{
	Oid			fc_lobjId_new;

	/*
	 * 创建一个具有空数据页的新大对象
	 */
	fc_lobjId_new = LargeObjectCreate(fc_lobjId);

	/*
	 * 对大对象所有者的依赖
	 *
	 * 我们在这里使用 LargeObjectRelationId 而不是
	 * LargeObjectMetadataRelationId 的原因是为了向
	 * 利用系统目录内部布局知识的应用程序提供向后兼容性。
	 * pg_largeobject_metadata 的 OID 和 pg_largeobject 的 loid
	 * 是相同的值，因此在这里没有实际差异。
	 */
	recordDependencyOnOwner(LargeObjectRelationId,
							fc_lobjId_new, GetUserId());

	/* 新大对象的创建后钩子 */
	InvokeObjectPostCreateHook(LargeObjectRelationId, fc_lobjId_new, 0);

	/*
	 * 提高命令计数器，以使新元组对后续操作可见。
	 */
	CommandCounterIncrement();

	return fc_lobjId_new;
}

/*
 *	inv_open -- 访问现有的大对象。
 *
 * 返回适当填充的大对象描述符。
 * 描述符及其附属数据在指定的内存上下文中分配，
 * 该上下文必须对调用者的目的足够持久。如果返回的描述符
 * 与之相关联的快照，调用者必须确保它的生命周期也足够长，
 * 例如，通过调用 RegisterSnapshotOnOwner
 */
LargeObjectDesc *
inv_open(Oid fc_lobjId, int fc_flags, MemoryContext fc_mcxt)
{
	LargeObjectDesc *fc_retval;
	Snapshot	fc_snapshot = NULL;
	int			fc_descflags = 0;

	/*
	 * 历史上，(INV_WRITE) 和 (INV_WRITE | INV_READ) 之间没有区别，
	 * 调用者在这两种情况下都可以读取大对象描述符。
	 */
	if (fc_flags & INV_WRITE)
		fc_descflags |= IFS_WRLOCK | IFS_RDLOCK;
	if (fc_flags & INV_READ)
		fc_descflags |= IFS_RDLOCK;

	if (fc_descflags == 0)
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
				 errmsg("invalid flags for opening a large object: %d",
						fc_flags)));

	/* 获取快照。如果请求写入，则使用瞬时快照。 */
	if (fc_descflags & IFS_WRLOCK)
		fc_snapshot = NULL;
	else
		fc_snapshot = GetActiveSnapshot();

	/* 在这里无法使用 LargeObjectExists，因为我们需要指定快照 */
	if (!fc_myLargeObjectExists(fc_lobjId, fc_snapshot))
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_OBJECT),
				 errmsg("large object %u does not exist", fc_lobjId)));

	/* 再次应用权限检查，指定快照 */
	if ((fc_descflags & IFS_RDLOCK) != 0)
	{
		if (!lo_compat_privileges &&
			pg_largeobject_aclcheck_snapshot(fc_lobjId,
											 GetUserId(),
											 ACL_SELECT,
											 fc_snapshot) != ACLCHECK_OK)
			ereport(ERROR,
					(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
					 errmsg("permission denied for large object %u",
							fc_lobjId)));
	}
	if ((fc_descflags & IFS_WRLOCK) != 0)
	{
		if (!lo_compat_privileges &&
			pg_largeobject_aclcheck_snapshot(fc_lobjId,
											 GetUserId(),
											 ACL_UPDATE,
											 fc_snapshot) != ACLCHECK_OK)
			ereport(ERROR,
					(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
					 errmsg("permission denied for large object %u",
							fc_lobjId)));
	}

	/* 可以创建描述符 */
	fc_retval = (LargeObjectDesc *) MemoryContextAlloc(fc_mcxt,
													sizeof(LargeObjectDesc));
	fc_retval->id = fc_lobjId;
	fc_retval->offset = 0;
	fc_retval->flags = fc_descflags;

	/* 调用者根据需要设置，在本文件中的函数中不使用 */
	fc_retval->subid = InvalidSubTransactionId;

	/*
	 * 快照（如果有的话）只是当前活动的快照。
	 * 如果需要，调用者将在其后替换为一个更持久的副本。
	 */
	fc_retval->snapshot = fc_snapshot;

	return fc_retval;
}

/*
 * 关闭之前通过 inv_open() 创建的大对象描述符，并
 * 释放其使用的长期内存。
 */
void inv_close(LargeObjectDesc *fc_obj_desc)
{
	Assert(PointerIsValid(fc_obj_desc));
	pfree(fc_obj_desc);
}

/*
 * 销毁现有的大对象（不要与描述符混淆！）
 *
 * 请注意，我们希望调用者已经进行过任何所需的权限检查。
 */
int inv_drop(Oid fc_lobjId)
{
	ObjectAddress fc_object;

	/*
	 * 删除大对象的任何评论和依赖关系
	 */
	fc_object.classId = LargeObjectRelationId;
	fc_object.objectId = fc_lobjId;
	fc_object.objectSubId = 0;
	performDeletion(&fc_object, DROP_CASCADE, 0);

	/*
	 * 提高命令计数器，以便后续事务中的大对象操作能够
	 * 看到元组的删除。
	 */
	CommandCounterIncrement();

	/* 出于历史原因，我们在成功时始终返回 1。 */
	return 1;
}

/*
 * 确定大对象的大小
 *
 * 注意：大对象可能会像 Unix 文件一样包含空隙。我们实际上返回的是
 * 最后一个字节的偏移量 + 1。
 */
static uint64 fc_inv_getsize(LargeObjectDesc *fc_obj_desc)
{
	uint64		fc_lastbyte = 0;
	ScanKeyData fc_skey[1];
	SysScanDesc fc_sd;
	HeapTuple	fc_tuple;

	Assert(PointerIsValid(fc_obj_desc));

	fc_open_lo_relation();

	ScanKeyInit(&fc_skey[0],
				Anum_pg_largeobject_loid,
				BTEqualStrategyNumber, F_OIDEQ,
				ObjectIdGetDatum(fc_obj_desc->id));

	fc_sd = systable_beginscan_ordered(lo_heap_r, lo_index_r,
									fc_obj_desc->snapshot, 1, fc_skey);

	/*
	 * 由于 pg_largeobject 索引在 loid 和 pageno 上，但我们
	 * 只限制 loid，因此反向扫描应以反向 pageno 顺序访问
	 * 大对象的所有页面。因此，检查第一个有效元组（== 最后一个有效页面）
	 * 就足够了。
	 */
	fc_tuple = systable_getnext_ordered(fc_sd, BackwardScanDirection);
	if (HeapTupleIsValid(fc_tuple))
	{
		Form_pg_largeobject fc_data;
		bytea	   *fc_datafield;
		int			fc_len;
		bool		fc_pfreeit;

		if (HeapTupleHasNulls(fc_tuple))	/* paranoia */
			elog(ERROR, "null field found in pg_largeobject");
		fc_data = (Form_pg_largeobject) GETSTRUCT(fc_tuple);
		fc_getdatafield(fc_data, &fc_datafield, &fc_len, &fc_pfreeit);
		fc_lastbyte = (uint64) fc_data->pageno * LOBLKSIZE + fc_len;
		if (fc_pfreeit)
			pfree(fc_datafield);
	}

	systable_endscan_ordered(fc_sd);

	return fc_lastbyte;
}

int64 inv_seek(LargeObjectDesc *fc_obj_desc, int64 fc_offset, int fc_whence)
{
	int64		fc_newoffset;

	Assert(PointerIsValid(fc_obj_desc));

	/*
	 * 如果您具有读取或写入权限，我们允许进行寻址/告知，因此
	 * 此处不需要进行权限检查。
	 */

	/*
	 * 注意：加法中可能发生溢出，但由于我们将拒绝
	 * 负结果，因此不需要额外测试。
	 */
	switch (fc_whence)
	{
		case SEEK_SET:
			fc_newoffset = fc_offset;
			break;
		case SEEK_CUR:
			fc_newoffset = fc_obj_desc->offset + fc_offset;
			break;
		case SEEK_END:
			fc_newoffset = fc_inv_getsize(fc_obj_desc) + fc_offset;
			break;
		default:
			ereport(ERROR,
					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
					 errmsg("invalid whence setting: %d", fc_whence)));
			fc_newoffset = 0;		/* 保持编译器安静 */
			break;
	}

	/*
	 * 在这里使用 errmsg_internal，因为我们不想在可翻译的字符串中暴露 INT64_FORMAT；
	 * 做得更好也不值得这个麻烦
	 */
	if (fc_newoffset < 0 || fc_newoffset > MAX_LARGE_OBJECT_SIZE)
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
				 errmsg_internal("invalid large object seek target: " INT64_FORMAT,
								 fc_newoffset)));

	fc_obj_desc->offset = fc_newoffset;
	return fc_newoffset;
}

int64 inv_tell(LargeObjectDesc *fc_obj_desc)
{
	Assert(PointerIsValid(fc_obj_desc));

	/*
	 * 如果您具有读取或写入权限，我们允许进行寻址/告知，因此
	 * 此处不需要进行权限检查。
	 */

	return fc_obj_desc->offset;
}

int inv_read(LargeObjectDesc *fc_obj_desc, char *fc_buf, int fc_nbytes)
{
	int			fc_nread = 0;
	int64		fc_n;
	int64		fc_off;
	int			fc_len;
	int32		fc_pageno = (int32) (fc_obj_desc->offset / LOBLKSIZE);
	uint64		fc_pageoff;
	ScanKeyData fc_skey[2];
	SysScanDesc fc_sd;
	HeapTuple	fc_tuple;

	Assert(PointerIsValid(fc_obj_desc));
	Assert(fc_buf != NULL);

	if ((fc_obj_desc->flags & IFS_RDLOCK) == 0)
		ereport(ERROR,
				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
				 errmsg("permission denied for large object %u",
						fc_obj_desc->id)));

	if (fc_nbytes <= 0)
		return 0;

	fc_open_lo_relation();

	ScanKeyInit(&fc_skey[0],
				Anum_pg_largeobject_loid,
				BTEqualStrategyNumber, F_OIDEQ,
				ObjectIdGetDatum(fc_obj_desc->id));

	ScanKeyInit(&fc_skey[1],
				Anum_pg_largeobject_pageno,
				BTGreaterEqualStrategyNumber, F_INT4GE,
				Int32GetDatum(fc_pageno));

	fc_sd = systable_beginscan_ordered(lo_heap_r, lo_index_r,
									fc_obj_desc->snapshot, 2, fc_skey);

	while ((fc_tuple = systable_getnext_ordered(fc_sd, ForwardScanDirection)) != NULL)
	{
		Form_pg_largeobject fc_data;
		bytea	   *fc_datafield;
		bool		fc_pfreeit;

		if (HeapTupleHasNulls(fc_tuple))	/* paranoia */
			elog(ERROR, "null field found in pg_largeobject");
		fc_data = (Form_pg_largeobject) GETSTRUCT(fc_tuple);

		/*
		 * 我们期望索引扫描将按顺序交付页面。然而，
		 * 如果 LO 包含未写入的“孔”，可能会缺少页面。我们希望缺失的部分以零的形式读取。
		 */
		fc_pageoff = ((uint64) fc_data->pageno) * LOBLKSIZE;
		if (fc_pageoff > fc_obj_desc->offset)
		{
			fc_n = fc_pageoff - fc_obj_desc->offset;
			fc_n = (fc_n <= (fc_nbytes - fc_nread)) ? fc_n : (fc_nbytes - fc_nread);
			MemSet(fc_buf + fc_nread, 0, fc_n);
			fc_nread += fc_n;
			fc_obj_desc->offset += fc_n;
		}

		if (fc_nread < fc_nbytes)
		{
			Assert(fc_obj_desc->offset >= fc_pageoff);
			fc_off = (int) (fc_obj_desc->offset - fc_pageoff);
			Assert(fc_off >= 0 && fc_off < LOBLKSIZE);

			fc_getdatafield(fc_data, &fc_datafield, &fc_len, &fc_pfreeit);
			if (fc_len > fc_off)
			{
				fc_n = fc_len - fc_off;
				fc_n = (fc_n <= (fc_nbytes - fc_nread)) ? fc_n : (fc_nbytes - fc_nread);
				memcpy(fc_buf + fc_nread, VARDATA(fc_datafield) + fc_off, fc_n);
				fc_nread += fc_n;
				fc_obj_desc->offset += fc_n;
			}
			if (fc_pfreeit)
				pfree(fc_datafield);
		}

		if (fc_nread >= fc_nbytes)
			break;
	}

	systable_endscan_ordered(fc_sd);

	return fc_nread;
}

int inv_write(LargeObjectDesc *fc_obj_desc, const char *fc_buf, int fc_nbytes)
{
	int			fc_nwritten = 0;
	int			fc_n;
	int			fc_off;
	int			fc_len;
	int32		fc_pageno = (int32) (fc_obj_desc->offset / LOBLKSIZE);
	ScanKeyData fc_skey[2];
	SysScanDesc fc_sd;
	HeapTuple	fc_oldtuple;
	Form_pg_largeobject fc_olddata;
	bool		fc_neednextpage;
	bytea	   *fc_datafield;
	bool		fc_pfreeit;
	union
	{
		bytea		hdr;
		/* 这是为了使联合足够大，以容纳 LO 数据块： */
		char		data[LOBLKSIZE + VARHDRSZ];
		/* 确保联合体的对齐方式足够好： */
		int32		align_it;
	}			fc_workbuf;
	char	   *fc_workb = VARDATA(&fc_workbuf.hdr);
	HeapTuple	fc_newtup;
	Datum		fc_values[Natts_pg_largeobject];
	bool		fc_nulls[Natts_pg_largeobject];
	bool		fc_replace[Natts_pg_largeobject];
	CatalogIndexState fc_indstate;

	Assert(PointerIsValid(fc_obj_desc));
	Assert(fc_buf != NULL);

	/* 强制可写性，因为快照否则可能是错误的 */
	if ((fc_obj_desc->flags & IFS_WRLOCK) == 0)
		ereport(ERROR,
				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
				 errmsg("permission denied for large object %u",
						fc_obj_desc->id)));

	if (fc_nbytes <= 0)
		return 0;

	/* 这个加法不会溢出，因为 nbytes 只有 int32 */
	if ((fc_nbytes + fc_obj_desc->offset) > MAX_LARGE_OBJECT_SIZE)
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
				 errmsg("invalid large object write request size: %d",
						fc_nbytes)));

	fc_open_lo_relation();

	fc_indstate = CatalogOpenIndexes(lo_heap_r);

	ScanKeyInit(&fc_skey[0],
				Anum_pg_largeobject_loid,
				BTEqualStrategyNumber, F_OIDEQ,
				ObjectIdGetDatum(fc_obj_desc->id));

	ScanKeyInit(&fc_skey[1],
				Anum_pg_largeobject_pageno,
				BTGreaterEqualStrategyNumber, F_INT4GE,
				Int32GetDatum(fc_pageno));

	fc_sd = systable_beginscan_ordered(lo_heap_r, lo_index_r,
									fc_obj_desc->snapshot, 2, fc_skey);

	fc_oldtuple = NULL;
	fc_olddata = NULL;
	fc_neednextpage = true;

	while (fc_nwritten < fc_nbytes)
	{
		/*
		 * 如果可能，获取 LO 的下一个现存页面。我们期望索引扫描
		 * 将这些按顺序交付---但可能会有孔。
		 */
		if (fc_neednextpage)
		{
			if ((fc_oldtuple = systable_getnext_ordered(fc_sd, ForwardScanDirection)) != NULL)
			{
				if (HeapTupleHasNulls(fc_oldtuple))	/* paranoia */
					elog(ERROR, "null field found in pg_largeobject");
				fc_olddata = (Form_pg_largeobject) GETSTRUCT(fc_oldtuple);
				Assert(fc_olddata->pageno >= fc_pageno);
			}
			fc_neednextpage = false;
		}

		/*
		 * 如果我们有一个现存的页面，看看它是否是我们想要写的页面，或是后面的页面。
		 */
		if (fc_olddata != NULL && fc_olddata->pageno == fc_pageno)
		{
			/*
			 * 用新数据更新现有页面。
			 *
			 * 首先，将旧数据加载到 workbuf 中
			 */
			fc_getdatafield(fc_olddata, &fc_datafield, &fc_len, &fc_pfreeit);
			memcpy(fc_workb, VARDATA(fc_datafield), fc_len);
			if (fc_pfreeit)
				pfree(fc_datafield);

			/*
			 * 填充任何孔
			 */
			fc_off = (int) (fc_obj_desc->offset % LOBLKSIZE);
			if (fc_off > fc_len)
				MemSet(fc_workb + fc_len, 0, fc_off - fc_len);

			/*
			 * 插入新数据的适当部分
			 */
			fc_n = LOBLKSIZE - fc_off;
			fc_n = (fc_n <= (fc_nbytes - fc_nwritten)) ? fc_n : (fc_nbytes - fc_nwritten);
			memcpy(fc_workb + fc_off, fc_buf + fc_nwritten, fc_n);
			fc_nwritten += fc_n;
			fc_obj_desc->offset += fc_n;
			fc_off += fc_n;
			/* 计算新页面的有效长度 */
			fc_len = (fc_len >= fc_off) ? fc_len : fc_off;
			SET_VARSIZE(&fc_workbuf.hdr, fc_len + VARHDRSZ);

			/*
			 * 形成并插入更新后的元组
			 */
			memset(fc_values, 0, sizeof(fc_values));
			memset(fc_nulls, false, sizeof(fc_nulls));
			memset(fc_replace, false, sizeof(fc_replace));
			fc_values[Anum_pg_largeobject_data - 1] = PointerGetDatum(&fc_workbuf);
			fc_replace[Anum_pg_largeobject_data - 1] = true;
			fc_newtup = heap_modify_tuple(fc_oldtuple, RelationGetDescr(lo_heap_r),
									   fc_values, fc_nulls, fc_replace);
			CatalogTupleUpdateWithInfo(lo_heap_r, &fc_newtup->t_self, fc_newtup,
									   fc_indstate);
			heap_freetuple(fc_newtup);

			/*
			 * 我们已经完成了这个旧页面。
			 */
			fc_oldtuple = NULL;
			fc_olddata = NULL;
			fc_neednextpage = true;
		}
		else
		{
			/*
			 * 写入全新页面。
			 *
			 * 首先，填充任何孔
			 */
			fc_off = (int) (fc_obj_desc->offset % LOBLKSIZE);
			if (fc_off > 0)
				MemSet(fc_workb, 0, fc_off);

			/*
			 * 插入新数据的适当部分
			 */
			fc_n = LOBLKSIZE - fc_off;
			fc_n = (fc_n <= (fc_nbytes - fc_nwritten)) ? fc_n : (fc_nbytes - fc_nwritten);
			memcpy(fc_workb + fc_off, fc_buf + fc_nwritten, fc_n);
			fc_nwritten += fc_n;
			fc_obj_desc->offset += fc_n;
			/* 计算新页面的有效长度 */
			fc_len = fc_off + fc_n;
			SET_VARSIZE(&fc_workbuf.hdr, fc_len + VARHDRSZ);

			/*
			 * 形成并插入更新后的元组
			 */
			memset(fc_values, 0, sizeof(fc_values));
			memset(fc_nulls, false, sizeof(fc_nulls));
			fc_values[Anum_pg_largeobject_loid - 1] = ObjectIdGetDatum(fc_obj_desc->id);
			fc_values[Anum_pg_largeobject_pageno - 1] = Int32GetDatum(fc_pageno);
			fc_values[Anum_pg_largeobject_data - 1] = PointerGetDatum(&fc_workbuf);
			fc_newtup = heap_form_tuple(lo_heap_r->rd_att, fc_values, fc_nulls);
			CatalogTupleInsertWithInfo(lo_heap_r, fc_newtup, fc_indstate);
			heap_freetuple(fc_newtup);
		}
		fc_pageno++;
	}

	systable_endscan_ordered(fc_sd);

	CatalogCloseIndexes(fc_indstate);

	/*
	 * 前进命令计数器，以便我的元组更新将在该事务中的后续
	 * 大对象操作中被看到。
	 */
	CommandCounterIncrement();

	return fc_nwritten;
}

void inv_truncate(LargeObjectDesc *fc_obj_desc, int64 fc_len)
{
	int32		fc_pageno = (int32) (fc_len / LOBLKSIZE);
	int32		fc_off;
	ScanKeyData fc_skey[2];
	SysScanDesc fc_sd;
	HeapTuple	fc_oldtuple;
	Form_pg_largeobject fc_olddata;
	union
	{
		bytea		hdr;
		/* 这是为了使联合足够大，以容纳 LO 数据块： */
		char		data[LOBLKSIZE + VARHDRSZ];
		/* 确保联合体的对齐方式足够好： */
		int32		align_it;
	}			fc_workbuf;
	char	   *fc_workb = VARDATA(&fc_workbuf.hdr);
	HeapTuple	fc_newtup;
	Datum		fc_values[Natts_pg_largeobject];
	bool		fc_nulls[Natts_pg_largeobject];
	bool		fc_replace[Natts_pg_largeobject];
	CatalogIndexState fc_indstate;

	Assert(PointerIsValid(fc_obj_desc));

	/* 强制可写性，因为快照否则可能是错误的 */
	if ((fc_obj_desc->flags & IFS_WRLOCK) == 0)
		ereport(ERROR,
				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
				 errmsg("permission denied for large object %u",
						fc_obj_desc->id)));

	/*
	 * 在这里使用 errmsg_internal，因为我们不想在可翻译的字符串中暴露 INT64_FORMAT；
	 * 做得更好也不值得这个麻烦
	 */
	if (fc_len < 0 || fc_len > MAX_LARGE_OBJECT_SIZE)
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
				 errmsg_internal("invalid large object truncation target: " INT64_FORMAT,
								 fc_len)));

	fc_open_lo_relation();

	fc_indstate = CatalogOpenIndexes(lo_heap_r);

	/*
	 * 设置以查找所有具有所需 loid 和 pageno >= target 的页面
	 */
	ScanKeyInit(&fc_skey[0],
				Anum_pg_largeobject_loid,
				BTEqualStrategyNumber, F_OIDEQ,
				ObjectIdGetDatum(fc_obj_desc->id));

	ScanKeyInit(&fc_skey[1],
				Anum_pg_largeobject_pageno,
				BTGreaterEqualStrategyNumber, F_INT4GE,
				Int32GetDatum(fc_pageno));

	fc_sd = systable_beginscan_ordered(lo_heap_r, lo_index_r,
									fc_obj_desc->snapshot, 2, fc_skey);

	/*
	 * 如果可能，获取截断点所在的页面。截断点
	 * 可能超出 LO 的末尾或在某个孔中。
	 */
	fc_olddata = NULL;
	if ((fc_oldtuple = systable_getnext_ordered(fc_sd, ForwardScanDirection)) != NULL)
	{
		if (HeapTupleHasNulls(fc_oldtuple))	/* paranoia */
			elog(ERROR, "null field found in pg_largeobject");
		fc_olddata = (Form_pg_largeobject) GETSTRUCT(fc_oldtuple);
		Assert(fc_olddata->pageno >= fc_pageno);
	}

	/*
	 * 如果我们找到了截断点的页面，我们需要在其中截断
	 * 数据。否则，如果我们在一个孔中，我们需要创建一个页面来
	 * 标记数据的结束。
	 */
	if (fc_olddata != NULL && fc_olddata->pageno == fc_pageno)
	{
		/* 首先，将旧数据加载到 workbuf 中 */
		bytea	   *fc_datafield;
		int			fc_pagelen;
		bool		fc_pfreeit;

		fc_getdatafield(fc_olddata, &fc_datafield, &fc_pagelen, &fc_pfreeit);
		memcpy(fc_workb, VARDATA(fc_datafield), fc_pagelen);
		if (fc_pfreeit)
			pfree(fc_datafield);

		/*
		 * 填充任何孔
		 */
		fc_off = fc_len % LOBLKSIZE;
		if (fc_off > fc_pagelen)
			MemSet(fc_workb + fc_pagelen, 0, fc_off - fc_pagelen);

		/* 计算新页面的长度 */
		SET_VARSIZE(&fc_workbuf.hdr, fc_off + VARHDRSZ);

		/*
		 * 形成并插入更新后的元组
		 */
		memset(fc_values, 0, sizeof(fc_values));
		memset(fc_nulls, false, sizeof(fc_nulls));
		memset(fc_replace, false, sizeof(fc_replace));
		fc_values[Anum_pg_largeobject_data - 1] = PointerGetDatum(&fc_workbuf);
		fc_replace[Anum_pg_largeobject_data - 1] = true;
		fc_newtup = heap_modify_tuple(fc_oldtuple, RelationGetDescr(lo_heap_r),
								   fc_values, fc_nulls, fc_replace);
		CatalogTupleUpdateWithInfo(lo_heap_r, &fc_newtup->t_self, fc_newtup,
								   fc_indstate);
		heap_freetuple(fc_newtup);
	}
	else
	{
		/*
		 * 如果我们找到的第一页面在截断点之后，我们就处于
		 * 一个孔中，我们将进行填充，但我们需要删除后面的页面
		 * 因为下面的循环不会再访问它。
		 */
		if (fc_olddata != NULL)
		{
			Assert(fc_olddata->pageno > fc_pageno);
			CatalogTupleDelete(lo_heap_r, &fc_oldtuple->t_self);
		}

		/*
		 * 写入全新页面。
		 *
		 * 填充到截断点的孔
		 */
		fc_off = fc_len % LOBLKSIZE;
		if (fc_off > 0)
			MemSet(fc_workb, 0, fc_off);

		/* 计算新页面的长度 */
		SET_VARSIZE(&fc_workbuf.hdr, fc_off + VARHDRSZ);

		/*
		 * 形成并插入新元组
		 */
		memset(fc_values, 0, sizeof(fc_values));
		memset(fc_nulls, false, sizeof(fc_nulls));
		fc_values[Anum_pg_largeobject_loid - 1] = ObjectIdGetDatum(fc_obj_desc->id);
		fc_values[Anum_pg_largeobject_pageno - 1] = Int32GetDatum(fc_pageno);
		fc_values[Anum_pg_largeobject_data - 1] = PointerGetDatum(&fc_workbuf);
		fc_newtup = heap_form_tuple(lo_heap_r->rd_att, fc_values, fc_nulls);
		CatalogTupleInsertWithInfo(lo_heap_r, fc_newtup, fc_indstate);
		heap_freetuple(fc_newtup);
	}

	/*
	 * 删除截断点之后的任何页面。如果初始搜索
	 * 没有找到页面，那么当然就没有更多要做的了。
	 */
	if (fc_olddata != NULL)
	{
		while ((fc_oldtuple = systable_getnext_ordered(fc_sd, ForwardScanDirection)) != NULL)
		{
			CatalogTupleDelete(lo_heap_r, &fc_oldtuple->t_self);
		}
	}

	systable_endscan_ordered(fc_sd);

	CatalogCloseIndexes(fc_indstate);

	/*
	 * 前进命令计数器，以便元组更新将被后续
	 * 大对象操作在该事务中看到。
	 */
	CommandCounterIncrement();
}
